package com.sample.springbatch.job.sample.config;

import com.sample.springbatch.config.BaseConfig;
import com.sample.springbatch.job.sample.service.SplitFlowStepService;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.FlowBuilder;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.step.tasklet.MethodInvokingTaskletAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;


/**
 * Created by zwg on 2017/7/14.
 */
//@Configuration
public class SplitFlowStepJobConfig extends BaseConfig {


    @Autowired
    SplitFlowStepService splitFlowStepService;

    @Bean(name = "splitFlowStepJob")
    public Job splitFlowStepJob() {
        return jobBuilderFactory.get("splitFlowStepJob")
                .incrementer(new RunIdIncrementer())
                //.flow(cleanQueueStep())
                .start(cleanQueueStep())  //和split 中的flow 会并行执行
                .split(new SimpleAsyncTaskExecutor())
                    .add(getPageDataFlow1(),getPageDataFlow2(),getPageDataFlow3())
                    .next(readQueueStep())
                    //.start(cleanQueueStep())
                    .build() //spilt end
                .build();
    }



    @Bean(name="getPageDataFlow1")
    public Flow getPageDataFlow1(){
        Step step = stepBuilderFactory.get("getPageDataStep1")
                .tasklet(getTasklet(1,20))
                .build();
        return new FlowBuilder<Flow>("getPageDataFlow1").start(step).build();
    }

    @Bean(name="getPageDataFlow2")
    public Flow getPageDataFlow2(){
        Step step = stepBuilderFactory.get("getPageDataStep2")
                .tasklet(getTasklet(21,40))
                .build();
        return new FlowBuilder<Flow>("getPageDataFlow2").start(step).build();
    }

    @Bean(name="getPageDataFlow3")
    public Flow getPageDataFlow3(){
        Step step = stepBuilderFactory.get("getPageDataStep3")
                .tasklet(getTasklet(41,60))
                .build();
        return new FlowBuilder<Flow>("getPageDataFlow3").start(step).build();
    }

    @Bean(name="cleanQueueStep")
    public Step cleanQueueStep(){
        MethodInvokingTaskletAdapter tasklet = new MethodInvokingTaskletAdapter();
        tasklet.setTargetObject(splitFlowStepService);
        tasklet.setTargetMethod("cleanQueue");

        return stepBuilderFactory.get("cleanQueueStep")
                .tasklet(tasklet)
                .build();
    }

    @Bean(name="readQueueStep")
    public Step readQueueStep(){
        MethodInvokingTaskletAdapter tasklet = new MethodInvokingTaskletAdapter();
        tasklet.setTargetObject(splitFlowStepService);
        tasklet.setTargetMethod("readQueue");

        return stepBuilderFactory.get("readQueueStep")
                .tasklet(tasklet)
                .build();
    }


    private MethodInvokingTaskletAdapter getTasklet(int startPage, int endPage){
        MethodInvokingTaskletAdapter tasklet = new MethodInvokingTaskletAdapter();
        tasklet.setTargetObject(splitFlowStepService);
        tasklet.setTargetMethod("getPageData");
        Object[] args = new Object[]{startPage,endPage};
        tasklet.setArguments(args);
        return tasklet;
    }
}
